package com.reactim.message.connection;

import com.reactim.message.config.ConnectionConfig;
import com.reactim.message.routing.MessageRoutingService;
import com.reactim.message.security.ConnectionAccessControl;
import com.reactim.message.security.ConnectionEncryption;
import com.reactim.message.security.ConnectionValidator;
import com.reactim.message.protection.ResourceProtectionService;
import com.reactim.message.metrics.ConnectionPerformanceMetrics;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis连接管理器实现
 * 使用Redis存储和管理用户连接信息，支持分布式环境
 */
@Service("distributedConnectionManager")
@Slf4j
@RequiredArgsConstructor
public class RedisConnectionManager implements ConnectionManager, BatchOperations {

    private final ReactiveRedisTemplate<String, ConnectionDetails> redisTemplate;
    private final ReactiveStringRedisTemplate stringRedisTemplate;
    private final RSocketRequesterHolder requesterHolder;
    private final ConnectionConfig connectionConfig;
    private final ConnectionAccessControl accessControl;
    private final ConnectionEncryption encryption;
    private final ConnectionValidator validator;
    private final MessageRoutingService messageRoutingService;
    private final FaultRecoveryService faultRecoveryService;
    private final ResourceProtectionService resourceProtectionService;
    private final ConnectionPerformanceMetrics performanceMetrics;
    private final String instanceId;

    /**
     * 存储用户连接信息（带安全验证和资源保护）
     *
     * @param userId 用户ID
     * @param requester RSocket请求器
     * @param connectionInfo 连接信息
     * @return 操作结果
     */
    @Override
    public Mono<Boolean> storeConnection(Long userId, RSocketRequester requester, ConnectionInfo connectionInfo) {
        // 获取客户端IP（从连接信息中获取，如果没有则使用默认值）
        String clientIp = connectionInfo.getClientIp() != null ? connectionInfo.getClientIp() : "127.0.0.1";
        
        // 记录操作开始时间
        long startTime = System.nanoTime();
        
        // 1. 检查资源保护限制
        return resourceProtectionService.checkConnectionAllowed(userId, clientIp)
                .flatMap(checkResult -> {
                    if (!checkResult.isAllowed()) {
                        log.warn("连接被资源保护拒绝: userId={}, ip={}, reason={}", 
                            userId, clientIp, checkResult.getReason());
                        return Mono.just(false);
                    }

                    // 2. 检查访问权限
                    return accessControl.checkConnectionPermission(userId)
                            .flatMap(accessResult -> {
                                if (!accessResult.isAllowed()) {
                                    log.warn("用户连接权限检查失败: userId={}, reason={}", userId, accessResult.getReason());
                                    return Mono.just(false);
                                }

                                // 3. 验证连接信息
                                return validator.validateConnection(userId, connectionInfo, clientIp)
                                        .flatMap(validationResult -> {
                                            if (!validationResult.isValid()) {
                                                log.warn("连接验证失败: userId={}, reason={}", userId, validationResult.getMessage());
                                                return Mono.just(false);
                                            }

                                            // 4. 执行实际的连接存储
                                            return performSecureConnectionStorage(userId, requester, connectionInfo, clientIp);
                                        });
                            });
                })
                .doOnSuccess(result -> {
                    // 记录性能指标
                    performanceMetrics.recordStoreOperation(() -> result);
                })
                .doOnError(error -> {
                    log.error("存储连接时发生错误: userId={}, error={}", userId, error.getMessage());
                })
                .onErrorResume(error -> {
                    log.error("存储连接时发生错误: userId={}, error={}", userId, error.getMessage());
                    return Mono.just(false);
                });
    }

    /**
     * 执行安全的连接存储
     */
    private Mono<Boolean> performSecureConnectionStorage(Long userId, RSocketRequester requester, ConnectionInfo connectionInfo, String clientIp) {
        // 设置实例ID
        connectionInfo.setInstanceId(instanceId);

        // 生成连接ID
        String connectionId = UUID.randomUUID().toString();

        // 存储RSocketRequester到本地缓存，并获取引用键
        String requesterKey = requesterHolder.storeRequester(userId, requester);

        // 加密敏感信息
        ConnectionInfo encryptedInfo = encryptSensitiveData(connectionInfo);

        // 创建连接详情
        ConnectionDetails details = ConnectionDetails.builder()
                .info(encryptedInfo)
                .requesterKey(requesterKey)
                .lastActiveTime(System.currentTimeMillis())
                .connectionId(connectionId)
                .build();

        // 获取Redis键
        String userConnectionKey = connectionConfig.getRedis().getUserConnectionKey(userId);
        String instanceConnectionsKey = connectionConfig.getRedis().getInstanceConnectionsKey(instanceId);
        String onlineUsersKey = connectionConfig.getRedis().getOnlineUsersKey();

        // 检查Redis健康状态，决定存储策略
        if (!faultRecoveryService.isRedisHealthy()) {
            // Redis不健康，使用故障转移存储
            return faultRecoveryService.storeConnectionInFailover(userId, details)
                    .doOnSuccess(success -> {
                        if (success) {
                            log.debug("用户连接信息已存储到故障转移存储: userId={}, instanceId={}", userId, instanceId);
                        }
                    });
        }

        // 存储连接信息到Redis
        return redisTemplate.opsForValue().set(
                        userConnectionKey,
                        details,
                        Duration.ofSeconds(connectionConfig.getRedis().getConnectionTtl())
                )
                // 将用户ID添加到实例连接集合
                .then(stringRedisTemplate.opsForSet().add(instanceConnectionsKey, userId.toString()))
                // 将用户ID添加到在线用户集合
                .then(stringRedisTemplate.opsForSet().add(onlineUsersKey, userId.toString()))
                // 增加用户连接计数
                .then(accessControl.incrementUserConnectionCount(userId))
                .map(result -> true)
                .doOnSuccess(success -> {
                    log.debug("用户连接信息存储成功: userId={}, instanceId={}", userId, instanceId);
                    // 记录连接建立到资源保护服务
                    resourceProtectionService.recordConnectionEstablished(userId, clientIp);
                })
                .doOnError(error -> {
                    log.error("用户连接信息存储失败: userId={}, error={}", userId, error.getMessage());
                    // Redis存储失败时，尝试故障转移存储
                    faultRecoveryService.storeConnectionInFailover(userId, details).subscribe();
                });
    }

    /**
     * 加密敏感数据
     */
    private ConnectionInfo encryptSensitiveData(ConnectionInfo connectionInfo) {
        if (!connectionConfig.getSecurity().isEnableEncryption()) {
            return connectionInfo;
        }

        try {
            ConnectionInfo encrypted = ConnectionInfo.builder()
                    .userId(connectionInfo.getUserId())
                    .deviceInfo(encryption.encrypt(connectionInfo.getDeviceInfo()))
                    .clientVersion(connectionInfo.getClientVersion())
                    .timestamp(connectionInfo.getTimestamp())
                    .instanceId(connectionInfo.getInstanceId())
                    .build();

            log.debug("连接信息加密完成: userId={}", connectionInfo.getUserId());
            return encrypted;

        } catch (Exception e) {
            log.error("连接信息加密失败: userId={}, error={}", connectionInfo.getUserId(), e.getMessage());
            return connectionInfo; // 加密失败时返回原始数据
        }
    }

    /**
     * 解密敏感数据
     */
    private ConnectionInfo decryptSensitiveData(ConnectionInfo connectionInfo) {
        if (!connectionConfig.getSecurity().isEnableEncryption() || connectionInfo == null) {
            return connectionInfo;
        }

        try {
            ConnectionInfo decrypted = ConnectionInfo.builder()
                    .userId(connectionInfo.getUserId())
                    .deviceInfo(encryption.decrypt(connectionInfo.getDeviceInfo()))
                    .clientVersion(connectionInfo.getClientVersion())
                    .timestamp(connectionInfo.getTimestamp())
                    .instanceId(connectionInfo.getInstanceId())
                    .build();

            log.debug("连接信息解密完成: userId={}", connectionInfo.getUserId());
            return decrypted;

        } catch (Exception e) {
            log.error("连接信息解密失败: userId={}, error={}", connectionInfo.getUserId(), e.getMessage());
            return connectionInfo; // 解密失败时返回原始数据
        }
    }

    /**
     * 获取用户连接信息
     *
     * @param userId 用户ID
     * @return 连接信息
     */
    @Override
    public Mono<ConnectionDetails> getConnection(Long userId) {
        return Mono.fromCallable(() -> performanceMetrics.recordRetrieveOperation(() -> {
            // 检查Redis健康状态
            if (!faultRecoveryService.isRedisHealthy()) {
                log.debug("Redis不健康，从故障转移存储获取连接: userId={}", userId);
                return faultRecoveryService.getConnectionFromFailover(userId)
                        .map(details -> {
                            if (details != null && details.getInfo() != null) {
                                // 解密敏感信息
                                ConnectionInfo decryptedInfo = decryptSensitiveData(details.getInfo());
                                details.setInfo(decryptedInfo);
                            }
                            return details;
                        }).block();
            }

            String userConnectionKey = connectionConfig.getRedis().getUserConnectionKey(userId);

            return redisTemplate.opsForValue().get(userConnectionKey)
                    .map(details -> {
                        if (details != null && details.getInfo() != null) {
                            // 解密敏感信息
                            ConnectionInfo decryptedInfo = decryptSensitiveData(details.getInfo());
                            details.setInfo(decryptedInfo);
                        }
                        return details;
                    })
                    .doOnSuccess(details -> {
                        if (details != null) {
                            log.debug("获取用户连接信息成功: userId={}", userId);
                        } else {
                            log.debug("用户连接信息不存在: userId={}", userId);
                        }
                    })
                    .doOnError(error -> {
                        log.error("获取用户连接信息失败: userId={}, error={}", userId, error.getMessage());
                        // Redis失败时，尝试从故障转移存储获取
                    })
                    .onErrorResume(error -> {
                        log.warn("从Redis获取连接失败，尝试故障转移存储: userId={}, error={}", userId, error.getMessage());
                        return faultRecoveryService.getConnectionFromFailover(userId)
                                .map(details -> {
                                    if (details != null && details.getInfo() != null) {
                                        // 解密敏感信息
                                        ConnectionInfo decryptedInfo = decryptSensitiveData(details.getInfo());
                                        details.setInfo(decryptedInfo);
                                    }
                                    return details;
                                });
                    }).block();
        }))
        .flatMap(result -> Mono.justOrEmpty(result));
    }

    /**
     * 移除用户连接信息
     *
     * @param userId 用户ID
     * @return 操作结果
     */
    @Override
    public Mono<Boolean> removeConnection(Long userId) {
        return getConnection(userId)
                .flatMap(details -> {
                    // 从本地缓存中移除RSocketRequester
                    requesterHolder.removeRequester(details.getRequesterKey());

                    // 检查Redis健康状态
                    if (!faultRecoveryService.isRedisHealthy()) {
                        // Redis不健康，从故障转移存储移除
                        return faultRecoveryService.removeConnectionFromFailover(userId)
                                .doOnSuccess(success -> {
                                    if (success) {
                                        log.debug("用户连接信息已从故障转移存储移除: userId={}", userId);
                                    }
                                });
                    }

                    // 获取Redis键
                    String userConnectionKey = connectionConfig.getRedis().getUserConnectionKey(userId);
                    String instanceConnectionsKey = connectionConfig.getRedis().getInstanceConnectionsKey(details.getInfo().getInstanceId());
                    String onlineUsersKey = connectionConfig.getRedis().getOnlineUsersKey();

                    // 从Redis中删除连接信息
                    return redisTemplate.delete(userConnectionKey)
                            // 从实例连接集合中移除用户ID
                            .then(stringRedisTemplate.opsForSet().remove(instanceConnectionsKey, userId.toString()))
                            // 从在线用户集合中移除用户ID
                            .then(stringRedisTemplate.opsForSet().remove(onlineUsersKey, userId.toString()))
                            // 减少用户连接计数
                            .then(accessControl.decrementUserConnectionCount(userId))
                            .map(removed -> true)
                            .doOnSuccess(success -> {
                                log.debug("用户连接信息移除成功: userId={}", userId);
                                // 记录连接断开到资源保护服务
                                String clientIp = details.getInfo().getClientIp() != null ? 
                                    details.getInfo().getClientIp() : "127.0.0.1";
                                resourceProtectionService.recordConnectionClosed(userId, clientIp);
                            })
                            .doOnError(error -> {
                                log.error("用户连接信息移除失败: userId={}, error={}", userId, error.getMessage());
                                // Redis移除失败时，也尝试从故障转移存储移除
                                faultRecoveryService.removeConnectionFromFailover(userId).subscribe();
                            });
                })
                .defaultIfEmpty(false);
    }

    /**
     * 检查用户是否在线
     *
     * @param userId 用户ID
     * @return 是否在线
     */
    @Override
    public Mono<Boolean> isUserOnline(Long userId) {
        String onlineUsersKey = connectionConfig.getRedis().getOnlineUsersKey();

        return stringRedisTemplate.opsForSet().isMember(onlineUsersKey, userId.toString())
                .defaultIfEmpty(false)
                .doOnError(error -> log.error("检查用户在线状态失败: userId={}, error={}", userId, error.getMessage()));
    }

    /**
     * 获取在线用户数量
     *
     * @return 在线用户数
     */
    @Override
    public Mono<Long> getOnlineUserCount() {
        String onlineUsersKey = connectionConfig.getRedis().getOnlineUsersKey();

        return stringRedisTemplate.opsForSet().size(onlineUsersKey)
                .defaultIfEmpty(0L)
                .doOnError(error -> log.error("获取在线用户数量失败: error={}", error.getMessage()));
    }

    /**
     * 向用户发送消息
     *
     * @param userId 用户ID
     * @param route 消息路由
     * @param message 消息数据
     * @return 发送结果
     */
    @Override
    public Mono<Void> sendMessageToUser(Long userId, String route, Object message) {
        return getConnection(userId)
                .flatMap(details -> {
                    // 检查连接是否在当前实例
                    if (instanceId.equals(details.getInfo().getInstanceId())) {
                        // 从本地缓存获取RSocketRequester
                        RSocketRequester requester = requesterHolder.getRequester(details.getRequesterKey());
                        if (requester != null) {
                            // 发送消息
                            return requester.route(route)
                                    .data(message)
                                    .send()
                                    .doOnSuccess(v -> log.debug("消息发送成功: userId={}, route={}", userId, route))
                                    .doOnError(error -> {
                                        log.error("消息发送失败: userId={}, route={}, error={}", userId, route, error.getMessage());
                                        // 连接可能已断开，移除连接信息
                                        removeConnection(userId).subscribe();
                                    });
                        } else {
                            log.warn("RSocketRequester不存在，可能已过期: userId={}, requesterKey={}", userId, details.getRequesterKey());
                            // 移除无效连接
                            return removeConnection(userId).then(Mono.empty());
                        }
                    } else {
                        // 跨实例消息发送，通过消息路由服务发送
                        log.debug("用户连接在其他实例，使用消息路由服务: userId={}, targetInstance={}, currentInstance={}",
                                userId, details.getInfo().getInstanceId(), instanceId);

                        return messageRoutingService.publishCrossInstanceMessage(
                                        details.getInfo().getInstanceId(),
                                        userId,
                                        route,
                                        message
                                )
                                .flatMap(success -> {
                                    if (success) {
                                        log.debug("跨实例消息路由成功: userId={}, targetInstance={}",
                                                userId, details.getInfo().getInstanceId());
                                    } else {
                                        log.warn("跨实例消息路由失败: userId={}, targetInstance={}",
                                                userId, details.getInfo().getInstanceId());
                                    }
                                    return Mono.<Void>empty();
                                });
                    }
                })
                .then()
                .onErrorResume(error -> {
                    log.error("向用户发送消息失败: userId={}, route={}, error={}", userId, route, error.getMessage());
                    return Mono.empty();
                });
    }

    /**
     * 更新用户最后活跃时间
     *
     * @param userId 用户ID
     * @return 操作结果
     */
    @Override
    public Mono<Boolean> updateLastActiveTime(Long userId) {
        return getConnection(userId)
                .flatMap(details -> {
                    // 更新最后活跃时间
                    details.setLastActiveTime(System.currentTimeMillis());

                    // 获取Redis键
                    String userConnectionKey = connectionConfig.getRedis().getUserConnectionKey(userId);

                    // 更新Redis中的连接信息
                    return redisTemplate.opsForValue().set(
                                    userConnectionKey,
                                    details,
                                    Duration.ofSeconds(connectionConfig.getRedis().getConnectionTtl())
                            )
                            .doOnSuccess(success -> log.debug("用户最后活跃时间更新成功: userId={}", userId))
                            .doOnError(error -> log.error("用户最后活跃时间更新失败: userId={}, error={}", userId, error.getMessage()));
                })
                .defaultIfEmpty(false);
    }

    /**
     * 获取指定实例的所有连接详情
     *
     * @param instanceId 实例ID
     * @return 连接详情列表
     */
    @Override
    public Mono<List<ConnectionDetails>> getConnectionsByInstance(String instanceId) {
        String instanceConnectionsKey = connectionConfig.getRedis().getInstanceConnectionsKey(instanceId);

        return stringRedisTemplate.opsForSet().members(instanceConnectionsKey)
                .map(userId -> Long.parseLong(userId))
                .flatMap(this::getConnection)
                .collectList()
                .doOnSuccess(connections -> log.debug("获取实例连接详情成功: instanceId={}, count={}", instanceId, connections.size()))
                .doOnError(error -> log.error("获取实例连接详情失败: instanceId={}, error={}", instanceId, error.getMessage()));
    }

    /**
     * 批量获取用户在线状态
     * 带故障恢复的批量操作
     *
     * @param userIds 用户ID列表
     * @return 用户ID到在线状态的映射
     */
    @Override
    public Mono<Map<Long, Boolean>> batchGetOnlineStatus(List<Long> userIds) {
        if (userIds == null || userIds.isEmpty()) {
            return Mono.just(new ConcurrentHashMap<>());
        }

        log.debug("批量获取用户在线状态: count={}", userIds.size());

        // 检查Redis健康状态
        if (!faultRecoveryService.isRedisHealthy()) {
            log.debug("Redis不健康，使用故障转移模式批量获取在线状态");
            return getOnlineStatusFromFailover(userIds);
        }

        String onlineUsersKey = connectionConfig.getRedis().getOnlineUsersKey();

        // 使用Flux并发查询替代批量操作
        return Flux.fromIterable(userIds)
                .flatMap(userId ->
                        stringRedisTemplate.opsForSet().isMember(onlineUsersKey, userId.toString())
                                .map(online -> Map.entry(userId, online))
                                .onErrorReturn(Map.entry(userId, false))
                )
                .collectMap(Map.Entry::getKey, Map.Entry::getValue)
                .doOnSuccess(result -> log.debug("批量获取用户在线状态成功: count={}", result.size()))
                .onErrorResume(error -> {
                    log.error("批量获取用户在线状态失败，尝试故障转移: error={}", error.getMessage());
                    return getOnlineStatusFromFailover(userIds);
                });
    }

    /**
     * 从故障转移存储获取在线状态
     */
    private Mono<Map<Long, Boolean>> getOnlineStatusFromFailover(List<Long> userIds) {
        return faultRecoveryService.getFailoverConnections()
                .map(failoverConnections -> {
                    Map<Long, Boolean> statusMap = new ConcurrentHashMap<>();
                    for (Long userId : userIds) {
                        statusMap.put(userId, failoverConnections.contains(userId));
                    }
                    return statusMap;
                });
    }

    /**
     * 批量获取连接详情
     * 带故障恢复的批量操作
     *
     * @param userIds 用户ID列表
     * @return 用户ID到连接详情的映射
     */
    @Override
    public Mono<Map<Long, ConnectionDetails>> batchGetConnections(List<Long> userIds) {
        if (userIds == null || userIds.isEmpty()) {
            return Mono.just(new ConcurrentHashMap<>());
        }

        log.debug("批量获取用户连接详情: count={}", userIds.size());

        // 检查Redis健康状态
        if (!faultRecoveryService.isRedisHealthy()) {
            log.debug("Redis不健康，使用故障转移模式批量获取连接详情");
            return getConnectionsFromFailover(userIds);
        }

        // 使用Flux并发查询替代批量操作
        return Flux.fromIterable(userIds)
                .flatMap(userId ->
                        getConnection(userId)
                                .map(details -> Map.entry(userId, details))
                                .onErrorReturn(Map.entry(userId, null))
                )
                .collectMap(Map.Entry::getKey, Map.Entry::getValue)
                .map(resultMap -> {
                    // 过滤掉null值
                    Map<Long, ConnectionDetails> filteredMap = new ConcurrentHashMap<>();
                    resultMap.forEach((userId, details) -> {
                        if (details != null) {
                            filteredMap.put(userId, details);
                        }
                    });
                    return filteredMap;
                })
                .doOnSuccess(result -> log.debug("批量获取用户连接详情成功: count={}", result.size()))
                .onErrorResume(error -> {
                    log.error("批量获取用户连接详情失败，尝试故障转移: error={}", error.getMessage());
                    return getConnectionsFromFailover(userIds);
                });
    }

    /**
     * 从故障转移存储获取连接详情
     */
    private Mono<Map<Long, ConnectionDetails>> getConnectionsFromFailover(List<Long> userIds) {
        return Flux.fromIterable(userIds)
                .flatMap(userId ->
                        faultRecoveryService.getConnectionFromFailover(userId)
                                .map(details -> Map.entry(userId, details))
                                .defaultIfEmpty(Map.entry(userId, null))
                )
                .collectMap(Map.Entry::getKey, Map.Entry::getValue)
                .map(resultMap -> {
                    // 过滤掉null值
                    Map<Long, ConnectionDetails> filteredMap = new ConcurrentHashMap<>();
                    resultMap.forEach((userId, details) -> {
                        if (details != null) {
                            filteredMap.put(userId, details);
                        }
                    });
                    return filteredMap;
                });
    }

    /**
     * 批量更新最后活跃时间
     * 使用Redis管道操作提高性能
     *
     * @param userIds 用户ID列表
     * @return 用户ID到更新结果的映射
     */
    @Override
    public Mono<Map<Long, Boolean>> batchUpdateLastActiveTime(List<Long> userIds) {
        if (userIds == null || userIds.isEmpty()) {
            return Mono.just(new ConcurrentHashMap<>());
        }

        log.debug("批量更新用户最后活跃时间: count={}", userIds.size());

        // 先批量获取连接详情
        return batchGetConnections(userIds)
                .flatMap(detailsMap -> {
                    long currentTime = System.currentTimeMillis();

                    // 使用并发更新替代批量操作
                    return Flux.fromIterable(detailsMap.entrySet())
                            .flatMap(entry -> {
                                Long userId = entry.getKey();
                                ConnectionDetails details = entry.getValue();

                                if (details != null) {
                                    // 更新最后活跃时间
                                    details.setLastActiveTime(currentTime);

                                    // 获取Redis键
                                    String userConnectionKey = connectionConfig.getRedis().getUserConnectionKey(userId);

                                    // 更新到Redis
                                    return redisTemplate.opsForValue().set(
                                                    userConnectionKey,
                                                    details,
                                                    Duration.ofSeconds(connectionConfig.getRedis().getConnectionTtl())
                                            )
                                            .map(result -> Map.entry(userId, true))
                                            .onErrorReturn(Map.entry(userId, false));
                                } else {
                                    return Mono.just(Map.entry(userId, false));
                                }
                            })
                            .collectMap(Map.Entry::getKey, Map.Entry::getValue);
                })
                .doOnSuccess(result -> log.debug("批量更新用户最后活跃时间成功: count={}", result.size()))
                .doOnError(error -> log.error("批量更新用户最后活跃时间失败: error={}", error.getMessage()))
                .onErrorReturn(new ConcurrentHashMap<>());
    }
}
